import java.io.*;
import java.net.*;
//import java.sql.Timestamp;
import java.util.*;
import common.*;

public class MessagePasser extends Thread
{
	public static ArrayList<String> partnerIPs;
	public static ArrayList<Integer> partnerPorts;
	//The default port for incoming connections to this MessagePasser
	public final int PORT;
	//The handler for the process that receives an incoming message
	private Process parent;
	//The socket that will listen for incoming connections
	private ServerSocket incoming;
	//A List of all nodes currently connected to this node
	public SocketThread [] connections;
	private boolean running;
	private boolean disconnected = false;
	protected static ArrayList<Integer> participants;
	protected static ArrayList<Integer> local_procs;
	protected static int network_size;
	static long log = Def.NOT_ASSIGNED;
	static long cnt = 0;
	
	/**
	* Constructor that builds a MessagePasser to listen for incoming connections and
	* pass messages on outgoing connections.
	* @param processMH A MessageHandler for the process using this MessagePasser, basicaly a callback function.
	* @param pID The Process ID of the process for this MessagePasser
	*/
	public MessagePasser(int _pID,Process _parent) throws IOException ////MessageHandler processMH, int pID) throws IOException
	{
		PORT = partnerPorts.get(_pID);
		parent = _parent;
		incoming = new ServerSocket(this.PORT);
		connections = new SocketThread [partnerIPs.size()];
		running = true;
		if(parent.myID == 0 || parent.myID == (partnerIPs.size() - 1))
		{
			Measure.initializeMessageMeasures();
		}
		this.start();
	}
	/**
	* Listens for incoming socket connections to this node, and build a SocketThread when
	* one is received.
	*/
	public void run()
	{
		while(running)
		{
			//Try to accept a new connection
			try
			{
				if( Conf.VERBOSE_POSTMAN )
					System.out.println("Waiting for new connection on port " + incoming.getLocalPort());
				Socket newConnection = incoming.accept();
				if( Conf.VERBOSE_POSTMAN )
					System.out.println("Accepted new connection");
				//Dispatch a Thread for this connection and add it to our list
				new SocketThread(newConnection);
			}
			catch(SocketException e)
			{
				//This is normal behavior when a connection is closed.
				System.out.println("ServerSocket closed");
			}
			//If we cannot accept, then punt
			catch(IOException e)
			{
				System.out.println("Error accepting incoming connection: " + e);
			}
		}
	}
	/**
	* Receive an incoming message, checking for internal status messages.
	* @param msg The incoming message from the other node
	*/
	public void receive( Message msg )
	{
		if( disconnected ) return;
		
		// notify received
		if( Conf.VERBOSE_POSTMAN )
			Utils.print_log( "-postman(" + parent.myID + "): a msg[t=" + msg.get_type() + "] is received from proc(" + msg.get_sender_id() + ")" );		
		
		// msg handlers
		int spinlock;
		switch(msg.get_type())
		{	
			case Def.DISCON:
				//This message represents separating this node from the network.
				Utils.print_log( "-postman(" + parent.myID + "): are you dumping me?" );
				killConnections();
				parent.get_shell().turnoff();
				parent.running = false;
				this.running = false;
				disconnected = true;
				
				break;
			
			case Def.TERMINATE:
				Utils.print_log( "*** TERMINATE action ***" );
				////********Move up to the end of run() in the future***********////
				//if(parent.myID == 0 || parent.myID == (partnerIPs.size() - 1))
				if( parent.captain )
				{
					System.out.println("The final OGM size was: " + Measure.ogm_log);
					System.out.println("The final ICM size was: " + Measure.icm_log);
				}
				////********Move up to the end of run() in the future***********////
				close();
				parent.get_shell().turnoff();
				parent.running = false;
				this.running = false;
				
				spinlock = 7000000;
				while( spinlock-- != 0 );
				System.exit( 0 );
				break;
				
			case Def.PING:
				break;
				
				
			case Def.KILL:
				parent.send( Def.DROP, Def.BROADCAST, null, null );
				spinlock = 9999999;
				while( spinlock-- != 0 );
				Utils.print_log( "-postman(" + parent.myID + "): you are dumping me." );
				killConnections();
				parent.get_shell().turnoff();
				parent.running = false;
				this.running = false;
				disconnected = true;
//				if( parent.captain ){
					//Utils.print_log( "-postman(" + parent.myID + "): next lowest = " + this.getLowestConnected() );
//				}
				
				break;
				
			case Def.DROP:
				if( msg.get_sender_id() == parent.myID )
					break;
				
				parent.holder = Def.NOT_ASSIGNED;
				
				if( connections[msg.get_sender_id()] != null ){
					connections[msg.get_sender_id()].close();
				}
				
				synchronized( participants ){
					if( participants.indexOf( msg.get_sender_id() ) != -1 ){
						participants.remove( participants.indexOf( msg.get_sender_id() ) );
						if( local_procs.indexOf( msg.get_sender_id() ) != -1 ){
							local_procs.remove( local_procs.indexOf( msg.get_sender_id() ) );
						}
						network_size = participants.size();
					}
					//Utils.print_log( "-postman: he's still in - so i got him" );
				}
				
				Utils.print_log( "-postman(" + parent.myID + "): proc(" + msg.get_sender_id() + ") is gone" );
				// remove him from routing table, update token tree (by process.java)
				
				if( Conf.ME_SCHEME == Def.ME_TOKEN ){
					Measure.incrimentICM(msg);
					parent.receive(msg);
				}
				break;
				
			case Def.JOIN:
				break;
				
			case Def.NOT_ASSIGNED:
				// dummy packet
				// shouldn't occur - I put this code for debugging purpose
				Utils.print_log( "ERROR: critical error (check your code @18rER#) -- MessagePasser.java" );
				break;
				
			default:
				Measure.incrimentICM(msg);
				parent.receive(msg);	// push the task should be handle at SW level
				break;
		}
	}
	/**
	* Send an outgoing message.
	* @param msg The outgoing message to Destination
	*/
	public boolean send(Message msg)
	{
		if( Conf.ME_SCHEME == Def.ME_TOKEN ){
			if( parent.holder == Def.NOT_ASSIGNED ){
				if( msg.get_type() == Def.REQUEST_READ || msg.get_type() == Def.REQUEST_WRITE || msg.get_type() == Def.REQUEST_TOKEN ){
					System.out.println( "-postman(" + parent.myID + "): the token tree is not yet built" );
					return false;
				}
			}
		}
		
		Measure.incrimentOGM(msg);
		if( msg.get_receiver_id() == Def.BROADCAST ){	// broadcasting
			for( int receiver = 0; receiver < partnerIPs.size(); receiver++ ){

				// Check to see that we have a connection to that node
				if(addConnection(receiver))
				{
					if( Conf.VERBOSE_POSTMAN )
						Utils.print_log( "-postman(" + parent.myID + "): a msg[t=" + msg.get_type() + "] is sent to proc(" + msg.get_receiver_id() + ")" );
					connections[receiver].send(msg);
				}
			}
			return true;
		}else{	// 1-to-1 comm
			int receiver = msg.get_receiver_id();
			if( receiver < 0 || Initializer.numNodes[1] <= receiver  ){
				if( Conf.VERBOSE_POSTMAN )
					Utils.print_log( "-postman(" + parent.myID + "): unknown proc(" + msg.get_receiver_id() + ")" );
				return false;
			}
			
			//Check to see that we have a connection to that node
			if(addConnection(receiver))
			{
				if( Conf.VERBOSE_POSTMAN )
					Utils.print_log( "-postman(" + parent.myID + "): a msg[t=" + msg.get_type() + "] is sent to proc(" + msg.get_receiver_id() + ")" );
				return connections[receiver].send(msg);
			}
		}
		return false;
	}//end-of-send(msg)
	/**
	 * In the ME_TOKENLESS scheme, the processes will call this method to see if one got granted from all participants. 
	 * return true the parameter id_list has every ID in the network, otherwise false.
	 */
	 //TODO Complete this !!!!
	public boolean is_this_everyone( ArrayList<Integer> id_list ){
		//Utils.print_log( "size(partner_list_mp) = " + partnerIPs.size() );
		//Utils.print_log( "size(got_GRANTd_from) = " + id_list.size() );
		
		//return (id_list.size() == partnerIPs.size());//// - 1));
		return (id_list.size() == participants.size());
	}
		
	/**
	* Tries to add a connection to this MessagePasser
	* @param ip The destination ip address of this connection
	* @param port The port to connect with
	*/
	public boolean addConnection(int partnerID)
	{
		//If we are already connected to partner, just let the machine know
		if(connections[partnerID] != null)
		{
			return connections[partnerID].isConnectionAlive();
		}
		//otherwise try to establish the new connection
		else
		{
			try
			{
				Socket s = new Socket(partnerIPs.get(partnerID), partnerPorts.get(partnerID));
				new SocketThread(s, partnerID);
				return true;
			}
			catch(IOException e)
			{	
				
				Utils.print_log("-network: Found a unreparible link" + partnerID);
				return false;
			}
		}
	}
	public void killConnections()
	{
		//Close all open connections
		for(SocketThread connection : connections)
		{
			if(connection != null)
			{
				connection.kill();
			}
			connection = null;
		}
	}
	/**
	* Close down this message passer
	*/
	public void close()
	{
		//Close all open connections
		killConnections();
		try
		{
			incoming.close();
		}
		catch(Exception e)
		{}
		//Stop accepting new connections
		running = false;
	}
	public static int getNetworkSize()
	{
		return partnerPorts.size();
	}
	public int getLowestConnected()
	{
		if( !participants.isEmpty() )
			return participants.get( 0 );
		else
			return -1;
		
//		for(int partner = 0; partner < partnerIPs.size(); partner++)
//		{
//			if(addConnection(partner))
//			{
//				return partner;
//			}
//		}
//		return -1;
	}
	public boolean isAlive(int partnerID)
	{
		return connections[partnerID].isAlive();
	}
	public String toString()
	{
		return Arrays.toString(connections);
	}
	
	/**
	* Inner class to handle all socket communication. Since multiple connections will
	* exist for each node, they each have to have multiple threads listening for incoming messages.
	**/
	private class SocketThread extends Thread
	{
		private Socket s;
		private boolean alive;
		private ObjectInputStream sin;
		private ObjectOutputStream sout;
		/**
		* Constructor that builds a thread to listen for incoming messages on this socket
		* @param s The new socket to run on
		*/
		public SocketThread(Socket s) throws IOException
		{
			initialize(s);
		}
		/**
		* Constructor that builds a thread to listen for incoming messages on this socket
		* @param s The new socket to run on
		* @param partnerID The ID of the partner we are building this socket for
		*/
		public SocketThread(Socket s, int partnerID) throws IOException
		{
			connections[partnerID] = this;
			initialize(s);
		}
		private void initialize(Socket s) throws IOException
		{
			this.s = s;
			sout = new ObjectOutputStream(s.getOutputStream());
			sin = new ObjectInputStream(s.getInputStream());
			alive = true;
			this.start();
		}
		/**
		* Constantly queries the TCP socket for incoming info, if there is nothing there
		* it blocks
		*/
		public void run()
		{
			////Message msg = null;
			while(alive)
			{
				Message msg = null;	// renew 'msg' on every loop
				
				try
				{
					msg = ( Message )sin.readObject();
					if( Conf.VERBOSE_POSTMAN )
						Utils.print_log( "-postman(" + parent.myID + "): a msg[t=" + msg.get_type() + "] is received from(" + msg.get_sender_id() + ")" );
				}
				catch(SocketException e)
				{
					
					//This is normal behavior when a connection is closed.
					//Utils.print_log("Socket Closed");
					//The Socket Closed, kill this connection
					kill();
				}
				catch(EOFException e)
				{
					if( System.currentTimeMillis() - log > 500 ){
						cnt = 0;
						log = System.currentTimeMillis();
					}
					
					cnt++;
					//System.err.println( cnt );
					
					if( cnt > local_procs.size() * ( participants.size() * .8 ) ){
						cnt = 0;
						System.err.println( "-postman: connection lost (physical)" );
						
						synchronized( participants ){
							for( int i = participants.size() - 1; i >= 0; i-- ){
								int temp = participants.get( i );
								if( local_procs.indexOf( temp ) == -1 ){
									if( connections[ temp ] != null ){
										connections[ temp ].close();
									}
									participants.remove( i );
									network_size = participants.size();
								}	
							}
						}
						parent.receive( new Message( Def.ANNOUNCE, -1, -1, System.currentTimeMillis(), null, null ) );
					}
//					
//					if( System.currentTimeMillis() - log < 1000 ){
//						parent.receive( new Message( Def.DROP, -1, -1, System.currentTimeMillis(), null, null ) );
//						log = System.currentTimeMillis();
//					}
//					

					//Utils.print_log("Connection lost");
					//The Socket Closed, kill this connection
					
					kill();
				}
				catch(java.io.OptionalDataException e){
					//e.printStackTrace();
					kill();
				}
				catch(java.io.StreamCorruptedException e){
					//e.printStackTrace();
					//The node hangs, kill this connection
					kill();
				}
				catch(IOException e)
				{
					//e.printStackTrace();
					Utils.print_log( "-postman(" + parent.myID + "): Error reading a new message: " + e);
				}
				catch(ClassNotFoundException e)
				{
					//e.printStackTrace();
					Utils.print_log( "-postman(" + parent.myID + "): Could not load Message Class, make sure Message.class is in this directory");
					System.exit(0);
				}
				catch(Exception e)
				{
					//e.printStackTrace();
					//Some other exception caused this socket to close.
					kill();
				}
				if(msg != null)
				{
					// a msg arrives to the port
					receive(msg);
				}
				else
				{
					//Utils.print_log("Got a null message.");
					if( Conf.VERBOSE_POSTMAN )
						Utils.print_log( "-postman(" + parent.myID + "): Detect disconnect" );
				}
			}
		}
		/**
		* Writes a message to the TCP socket.
		* @param msg The outgoing message to place on the socket
		*/
		public boolean send(Message msg)
		{
			try
			{
				sout.writeObject(msg);
			}
			catch(Exception e)
			{
				return false;
			}
			return true;
		}
		
		/**
		* Close down this socket and the listening Thread
		*/
		public void close()
		{
			try
			{
				s.close();
			}
			catch(Exception e)
			{
			}
		}
		/**
		* Returns whether or not this connection (Node) is still alive.
		* @return true if this node is alive.
		*/
		public boolean isConnectionAlive()
		{
			return alive;
		}
		public void kill()
		{
			if( Conf.VERBOSE_POSTMAN )
				Utils.print_log( "-postman(" + parent.myID + "): remove the node from my routing table" );
			alive = false;
			//close();
		}
		/**
		* @return a string representation of this node.
		*/
		public String toString()
		{
			return s.toString();
		}
	}
}

/* Todo
 * - handling of disconnected node (addConnection should be attempted for these nodes)
 * - the msg handling routine - Ping / drop / kill stuff
 * - Complete is_this_everyone()
 */